#ifndef __M_REACTOR_H__
#define __M_REACTOR_H__
#include <iostream>
#include <fstream>
#include <vector>
#include <string>
#include <list>
#include <unordered_set>
#include <unordered_map>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <functional>
#include <memory>
#include <cassert>
#include <cstring>
#include <cerrno>
#include <chrono>
#include <unistd.h>
#include <fcntl.h>
#include <signal.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <sys/timerfd.h>
#include <sys/eventfd.h>

namespace bit_net {
    #define INF 0
    #define DBG 1
    #define ERR 2
    #define DEFAULT_LOG_LEVEL ERR
    #define LOG(level, format, ...) {\
        if (level >= DEFAULT_LOG_LEVEL) {\
            time_t t = time(NULL);\
            struct tm *m = localtime(&t);\
            char *ts = asctime(m);\
            ts[strlen(ts) - 1] = '\0';\
            fprintf(stdout, "[%s:%d %s %p] " format "\n", __FILE__, __LINE__, ts, pthread_self(), ##__VA_ARGS__);\
        }\
    }
    #define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__);
    #define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__);
    #define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__);

    #define DEFAULT_BUFFER_SIZE 1024
    #define MAX_LISTEN 1024
    #define MAX_EPOLLEVENTS 512
    #define DEFAULT_TIMEOUT 10

    class Buffer;
    class Socket;
    class Channel;
    class Connection;
    class EventLoop;
    class Poller;
    class TimerQueue;

    using ConnectionPtr = std::shared_ptr<Connection>;
    using ConnectionComplateCallback = std::function<void(const ConnectionPtr&)>;
    using ConnectionEventCallback = std::function<void(const ConnectionPtr&)>;
    using ConnectionMsgCallback = std::function<void(const ConnectionPtr&, Buffer *)>;
    using ConnectionCloseCallback = std::function<void(const ConnectionPtr&)>;
    using OnTimerCallback = std::function<void()>;
    
    enum State {DISCONNECTED, CONNECTING, CONNECTED, DISCONNECTING};

    class Buffer {
        private:
            std::vector<char> _buffer;
            size_t _reader_idx;
            size_t _writer_idx;
        private:
            /*返回缓冲区起始地址*/
            char *begin();
            /*返回当前读取位置之前空闲空间大小*/
            size_t prepend_size();
            /*查看剩余空间是否足够len大小，不够就resize空间大小，够了就将当前读位置的数据移动到空间起始处*/
            void make_space(size_t len);
            /*将当前写位置向后移动*/
            void has_written(size_t len);
            /*确保剩余可写空间大小足够len长度，不够的话就make_space*/
            void ensure_write_able_size(size_t len);
            void retrieve_all();
            /*返回当前写位置地址*/
            char *begin_write();
        public:
            Buffer(size_t init_size);
            /*返回当前可读数据大小*/
            size_t readable_size();
            /*返回当前可读数据后的可写空间大小*/
            size_t writeable_size();
            /*返回当前读取位置地址*/
            const char *peek();
            /*返回当前行的结束地址*/
            const char *find_eol();
            /*当前读位置向后偏移指定长度*/
            void retrieve(size_t size);
            /*将指定长度的可读数据以string取出，并向后偏移读位置*/
            std::string retrieve_as_string(size_t len);
            std::string retrieve_all_as_string();
            std::string getline();
            /*将数据写入缓冲区中*/
            void append(const char *data, size_t len);
    };
    class Socket {
        private:
            int _sockfd;
        public:
            Socket();
            Socket(int fd);
            ~Socket();
            int fd();
            void set_reuse();
            void set_nonblock();
            bool socket();
            bool bind(const std::string &ip, int port);
            bool connect(const std::string &ip, int port);
            bool listen(int backlog);
            int accept();
            bool create_listener(int port, const std::string &ip);
            bool create_client(const std::string &srvip, int srvport);
            ssize_t recv(void *buf, size_t len, int flag);
            ssize_t nonblock_recv(void *buf, size_t len);
            ssize_t send(const void *data, size_t len, int flag);
            ssize_t nonblock_send(const void *buf, size_t len);
            void close();
    };
    class Channel {
        private:
            using EventCallback = std::function<void()>;
            int _fd;
            EventLoop *_loop;
            uint32_t _events;
            uint32_t _revents;
            EventCallback _event_callback;/*可读事件回调*/
            EventCallback _read_callback;/*可读事件回调*/
            EventCallback _write_callback;/*可写事件回调*/
            EventCallback _error_callback;/*错误事件回调*/
            EventCallback _close_callback;/*关闭事件回调*/
        public:
            Channel(EventLoop *loop, int fd);
            int fd();
            uint32_t events();
            EventLoop *owner_loop();
            /*由Connection设置的设置不同的事件回调处理函数*/
            void set_event_callback(const EventCallback &cb);
            void set_read_callback(const EventCallback &cb);
            void set_write_callback(const EventCallback &cb);
            void set_error_callback(const EventCallback &cb);
            void set_close_callback(const EventCallback &cb);
            /*调用eventloop的remove_channel移除当前channel的事件监控*/
            void remove();
            /*调用eventloop的update_channel修改当前channel的事件监控*/
            void update();
            /*由poller哪里针对当前描述符就绪的事件进行revents的设置*/
            void set_revents(uint32_t events);
            /*将channel添加到eventloop的管理中，并启动监听可读事件*/
            void enable_read();
            /*移除当前channel的可读事件监控*/
            void disable_read();
            /*将channel添加到eventloop的管理中，并启动监听可读事件*/
            void enable_write();
            /*移除当前channel的可写事件监控*/
            void disable_write();
            /*移除当前channel的所有事件监控*/
            void disable_all();
            /*判断当前channel是否进行了可读事件的监控*/
            bool readable();
            /*判断当前channel是否进行了可写事件的监控*/
            bool writeable();
            /*提供给eventloop的唯一事件回调函数--内部根据就绪的不同事件调用Connection设置的不同回调*/
            void handle_event();
    };
    /*首先Any类肯定不能是一个模板类，否则编译的时候 Any<int> a,  Any<float>b,需要传类型作为模板参数，也就是说在使用的时候就要确定其类型*/
    /*这是行不通的，因为保存在Content中的协议上下文，我们在定义any对象的时候是不知道他们的协议类型的，因此无法传递类型作为模板参数*/
    /*因此考虑Any内部设计一个模板容器holder类，可以保存各种类型数据*/
    /*而因为在Any类中无法定义这个holder对象或指针，因为any也不知道这个类要保存什么类型的数据，因此无法传递类型参数*/
    /*所以，定义一个基类placehoder，让holder继承于placeholde，而Any类保存父类指针即可*/
    /*当需要保存数据时，则new一个带有模板参数的子类holder对象出来保存数据，然后让Any类中的父类指针，指向这个子类对象就搞定了*/
    /*当前的Any类只是摘出了一部分进行实现，并没有完全的进行各项完善，但是已经够当前的使用了*/
    class Any {
        public:
            Any():_content(NULL) {}
            /*为了能够接收所有类型的对象，因此将构造函数定义为一个模板函数*/
            template<typename T>
            Any(const T &val) : _content(new holder<T>(val)){}
            Any(const Any &other): _content(other._content?other._content->clone() : NULL) {}
            ~Any() { if (_content) delete _content; }
            const std::type_info &type() { return _content ? _content->type() : typeid(void); }
            Any& swap(Any &other) {
                std::swap(_content, other._content);
                return *this;
            }
            template<typename T>
            T* get() {
                assert(typeid(T) == _content->type());
                return &((holder<T>*)_content)->val;
            }
            template<typename T>
            Any& operator=(const T &val) {
                /*为val构建一个临时对象出来，然后进行交换，这样临时对象销毁的时候，顺带原先保存的placeholder也会被销毁*/
                Any(val).swap(*this);
                return *this;
            }
            Any& operator=(Any other) {
                /*这里要注意形参只是一个临时对象，进行交换后就会释放，所以交换后，原先保存的placeholder指针也会被销毁*/
                other.swap(*this);
                return *this;
            }
            
        private:
            class placeholder {
                public:
                    virtual ~placeholder() {}
                    virtual const std::type_info &type()  = 0;
                    virtual placeholder *clone() = 0;
            };
            template <typename T>
            class holder : public placeholder {
                public:
                    holder(const T &v):val(v) {}
                    ~holder() {}
                    const std::type_info &type() { return typeid(T); }
                    placeholder *clone() { return new holder(val); }
                public:
                    T val;
            };
            placeholder *_content;
    };
    class Connection: public std::enable_shared_from_this<Connection> {
        private:
            uint64_t _id; 
            EventLoop *_loop;
            std::unique_ptr<Socket> _sock;
            std::unique_ptr<Channel> _channel;
            State _state;
            Buffer _in_buffer;
            Buffer _out_buffer;
            Any _context;
        private:
            /*由外界设置的回调函数，对外的接口都用的是shared_ptr，防止Connection被释放后操作出错*/
            ConnectionComplateCallback _connected_callback; //连接建立完毕后的回调函数，主要用于不同协议设置context上下文信息
            ConnectionEventCallback _event_callback;;//只要有事件到来，不管是什么事件都要进行的操作--目前主要用于刷新定时任务，避免被判定非活跃
            ConnectionMsgCallback _message_callback; // 有消息到来后的回调函数，不同的协议对in_buffer数据的不同处理
            ConnectionCloseCallback _close_callback; //关闭时最后的回调
        private:
            /*连接建立成功后的处理函数：主要是给channel设置回调，以及添加定时器销毁任务*/
            void establisehd_in_loop();
            /*这个是只进行事件监控的移除和描述符的关闭，并不释放Connection资源*/
            void shutdown_in_loop();
            /*将数据放到发送缓冲区的接口*/
            void send_in_loop(const void *msg, size_t len);
            /*给Channel设置的五种事件回调函数*/
            void handle_event();
            void handle_read();
            void handle_write();
            void handle_error();
            void handle_close();
        public:
            Connection(EventLoop *loop, int fd, uint64_t _id);
            ~Connection();
            EventLoop *get_loop();
            int id();
            int fd();
            bool connected();
            Buffer *input_buffer();
            Buffer *output_buffer();
            void set_context(const Any &context);
            Any *get_context();
            /*connected是连接完成后进行的操作--在TcpServer中设置，主要是设置协议处理的上下文*/
            void set_connected_callback(const ConnectionComplateCallback &cb);
            /*event是只要有事件被触发，就调用的回调函数--在TcpServer中设置，主要是刷新定时任务，避免被判定为非活跃连接*/
            void set_event_callback(const ConnectionEventCallback &cb);
            /*message是将数据读取到接收缓冲区后进行操作*/
            void set_on_message_callback(const ConnectionMsgCallback &cb);
            /*close是进行连接关闭时进行的操作--内部主要是移除TcpServer中保存的Connection信息，然后移除监控与关闭描述符*/
            void set_close_callback(const ConnectionCloseCallback &cb);
            /*向外提供的数据发送接口，实际上是把数据放到发送缓冲区，然后启动可读事件监控*/
            void send(const void *msg, size_t len);
            /*向外提供的关闭接口，只进行事件监控的移除和描述符的关闭，并不进行Connection资源的释放，资源等超时自动进行释放*/
            void shutdown();
            /*这个函数是提供给TcpServer中使用，新连接创建成功后压入eventloop的连接完成后的处理函数，主要功能是设置Connection状态，以及添加可读事件监控，然后调用外部设置的连接完成回调*/
            void conn_established();
    };
    class Poller {
        private:
            int _epfd;
            struct epoll_event _evs[MAX_EPOLLEVENTS];
            std::unordered_map<int, Channel*> _channels;
            EventLoop *_loop;
        private:
            void update(int op, Channel *channel);
        public:
            Poller(EventLoop *loop);
            bool has_channel(Channel *channel);
            void update_channel(Channel *channel);
            void remove_channel(Channel *channel);
            bool poll(std::vector<Channel*> *active_channels, int timeout);
    };
    class Timer {
        private:
            using ReleaseCallback = std::function<void()>;
            uint64_t _timer_id;
            OnTimerCallback _timer_callback;
            ReleaseCallback _release_callback;
        public:
            Timer(uint64_t timer_id): _timer_id(timer_id){}
            ~Timer(){ 
                /*执行定时任务*/
                _timer_callback(); 
                /*由timequeue设置的释放回调，主要是移除timequeue中保存的Timer信息*/
                _release_callback(); 
            }
            uint64_t id() { return _timer_id; }
            /*这个是设置当前的定时任务到期时要执行的任务函数*/
            void set_on_time_callback(const OnTimerCallback &cb) {_timer_callback = cb; }
            /*这个是当前的定时任务到期时，从timer_queue的hash表中移除任务信息*/
            void set_release_callback(const ReleaseCallback &cb) {_release_callback = cb; }
    };

    class TimerQueue {
        public:
            using WeakTimer = std::weak_ptr<Timer>;
            using PtrTimer = std::shared_ptr<Timer>;
            using Bucket = std::unordered_set<PtrTimer>;
            using BucketList = std::list<Bucket>;
            TimerQueue(EventLoop *loop);
            /*添加一个定时任务*/
            void timer_add(const OnTimerCallback &cb, uint64_t id);
            /*根据id刷新定时任务*/
            void timer_refresh(uint64_t id);
            /*设置给Timer，最终定时任务执行完毕后从timequeue移除timer信息的回调函数*/
            void erase(uint64_t id) ;
            /*这是系统定时器每次时间到了要进行的操作：主要是读取定时器数据，然后所有定时任务向前走一步*/
            /*而被pop出来的shared_ptr计数为0的任务就都是时间到了要执行的任务*/
            void on_time();
        private:
            /*创建并启动系统定时器*/
            static int create_timerfd();
            /*读取定时器数据函数*/
            void read_timerfd();
        private:
            EventLoop *_loop;
            int _timer_fd;
            std::unique_ptr<Channel> _channel;

            int _capacity;
            std::unordered_map<uint64_t, WeakTimer> _timers;
            BucketList _conns;
    };
    class Acceptor {
        private:
            EventLoop *_loop;
            Socket _sock;
            Channel *_channel;
            using NewConnectionCallback = std::function<void(int)>;
            NewConnectionCallback _new_conn_callback;/*这个回调函数功能是等获取到新连接之后， 给新连接创建Connection结构设置各项数据的接口*/
        public:
            Acceptor(EventLoop *loop, int port);
            void set_new_conn_callback(const NewConnectionCallback &cb);
            /*这个函数是accept获取新连接的函数，到时候会将这个函数设置为channel的read_callback*/
            void read_handler();
    };
    typedef enum { MAIN_LOOP, SUB_LOOP, }LoopType;
    class EventLoop {
        private:
            using Functor = std::function<void()>;
            LoopType _loop_type;
            std::thread::id _thread_id;
            /*因为有可能当前eventloop线程卡在等待描述符就绪事件上，因此通过给weakupfd写入数据唤醒阻塞*/
            int _weakup_fd;
            int _timeout;
            Poller *_poller;
            Channel *_weakup_channel;
            TimerQueue *_timer_queue;

            std::mutex _mutex;
            std::vector<Functor> _pending_functor;
        private:
            static int create_event_fd();
            /*因为线程有可能阻塞在等待描述符事件就绪上，因此创建一个weakup_fd添加到监控中用于唤醒阻塞*/
            /*唤醒操作说白了也就是给weakup_fd中写入个数据就行*/
            void weakup();
            /*读取weakup_fd中的数据*/
            void weakup_channel_handler_read();
            /*对任务队列中的任务进行处理*/
            void do_pending_functor();
        public:
            EventLoop(LoopType loop_type);
            bool is_main() {return _loop_type == MAIN_LOOP; }
            /*启动事件监控，并在每次监控事件处理完毕后，去处理任务队列中的任务*/
            void start();
            /*如果当前就是在eventloop线程中，则直接调用执行，否则就压入到任务队列中*/
            void run_in_loop(Functor cb);
            /*将需要执行的函数压入eventloop线程任务队列中，然后如果当前不在eventloop线程中，则唤醒eventloop线程*/
            void queue_in_loop(Functor cb);
            /*判断当前运行流程是否运行于eventloop线程内*/
            bool is_in_loop_thread() ;
            /*在poller中更新channel对应的监控事件*/
            void update_channel(Channel *channel);
            /*从poller中移除channel对应fd的事件监控*/
            void remove_channel(Channel *channel);
            /*判断poller中是否已经添加了channel*/
            bool has_channel(Channel *channel);
            /*添加一个新的定时任务*/
            void timer_task_enqueue(const OnTimerCallback &cb, uint64_t timer_id);
            /*刷新定时任务，或者说重新对指定定时任务进行计时*/
            void timer_refresh(uint64_t timer_id);
    };

    /*把eventloop和thread放一起，保证在线程内实例化eventloop才能确保eventloop的thread_id成员是对应线程ID*/
    class LoopThread {
        private:
            EventLoop *_loop;
            std::mutex _mutex;
            std::condition_variable _cond;
            std::thread _thread;
        private:
            /*线程入口函数，线程内实例化eventloop对象，并启动*/
            void thread_entry();
        public:
            LoopThread();
            /*如果外界想要获取线程对应的eventloop，则阻塞至_loop不为NULL*/
            EventLoop *own_loop();
    };


    class EventLoopThreadPool {
        private:
            int _num_thread;
            int _next;
            EventLoop *_base_loop;
            std::vector<LoopThread*> _threads;
            std::vector<EventLoop*> _loops;
        public:
            EventLoopThreadPool(EventLoop *baseloop);
            void ser_thread_count(int num);
            /*每个新连接都会分配给一个eventloop线程，这里采用RR轮转分配*/
            EventLoop *next_loop();
            /*根据线程数量创建LoopThread，并在当前主线程中启动base_loop运行*/
            void start();
    };
    void connected_default_callback(const ConnectionPtr &conn) {
        DBG_LOG("new Connection:%p, socket:%d", conn.get(), conn->fd());
    }
    void onmessage_default_callback(const ConnectionPtr &conn, Buffer *buffer) {
        std::string str = buffer->retrieve_all_as_string();
        INF_LOG("%s", str.c_str());
    }
    class TcpServer {
        public:
            TcpServer(EventLoop *loop, int port):
                _port(port), 
                _next_id(1),
                _base_loop(loop),
                _acceptor(new Acceptor(loop, port)), 
                _thread_pool(new EventLoopThreadPool(loop)),
                _connected_callback(connected_default_callback),
                _on_message_callback(onmessage_default_callback){
                /*给监听套接字对应Connection对象设置获取新连接成功后的处理函数，主要是给新连接申请Connection对象， 设置各项参数*/
                _acceptor->set_new_conn_callback(std::bind(&TcpServer::new_connection, this, std::placeholders::_1));
            }
            void set_connected_callback(const ConnectionComplateCallback &cb) { _connected_callback = cb; }
            void set_message_callback(const ConnectionMsgCallback &cb) { _on_message_callback = cb; }
            /*设置线程数量，默认0个IO线程--也就是将监听与IO放在一起*/
            void set_thread_count(int nums) {
                assert(nums >= 0);
                _thread_pool->ser_thread_count(nums);
            }
            /*启动eventloop线程池，并且压入第一个需要处理的函数---将监听套接字添加到Poller事件监控*/
            void start() {
                _thread_pool->start();
            }
        private:
            /*获取新连接成功后的回调--主要是申请Connection对象，设置关闭处理函数，然后将连接建立完成函数压入eventloop处理队列中*/
            void new_connection(int sockfd) {
                EventLoop *ioloop = _thread_pool->next_loop();
                ConnectionPtr conn(new Connection(ioloop, sockfd, _next_id));
                /*连接建立完成后的回调函数--主要用于协议这里的上下文设置*/
                conn->set_connected_callback(_connected_callback);
                /*接收数据后的处理回调函数--主要用于协议这里的数据处理函数*/
                conn->set_on_message_callback(_on_message_callback);
                /*事件回调函数，只要有事件，不管什么事件都执行的操作--主要是刷新连接的定时任务，避免被判定为非活跃连接*/
                conn->set_event_callback(std::bind(&TcpServer::refresh_timer_task, this, std::placeholders::_1));
                /*给当前的connction添加一个超时任务，也就是非活跃超时后销毁资源*/
                /*这里需要注意，添加定时任务一定要在添事件监控之前，否则有可能出现读事件发生了，但是定时任务还没添加，刷新定时器任务的时候就会出问题*/
                ioloop->run_in_loop(std::bind(&TcpServer::new_timer_task, this, conn));
                /*这里可以理解就是调用conn_established函数，不过不是当前调用，而是压入到eventloop的任务队列中，在io线程中调用，进行连接建立完毕后的下一步操作*/
                ioloop->run_in_loop(std::bind(&Connection::conn_established, conn));
                _connections.insert(std::make_pair(_next_id, conn));
                _next_id++;
            }
            /*刷新定时任务*/
            void refresh_timer_task(const ConnectionPtr &conn) {
                EventLoop *ioloop = conn->get_loop();
                ioloop->run_in_loop(std::bind(&TcpServer::refresh_timer_task_in_loop, this, conn));
            }
            void refresh_timer_task_in_loop(const ConnectionPtr &conn) {
                EventLoop *ioloop = conn->get_loop();
                ioloop->timer_refresh(conn->id());
            }
            void new_timer_task(const ConnectionPtr &conn) {
                EventLoop *ioloop = conn->get_loop();
                /*将连接的从TcpServer中移除作为定时任务，添加到定时任务队列中*/
                ioloop->timer_task_enqueue(std::bind(&TcpServer::remove_inactive_connection, this, conn), conn->id());
            }
            /*移除超时非活跃连接*/
            void remove_inactive_connection(const ConnectionPtr &conn) {
                EventLoop *ioloop = conn->get_loop();
                ioloop->run_in_loop(std::bind(&TcpServer::remove_inactive_connection_in_loop, this, conn));
            }
            /*这个函数是在eventloop线程中被调用的， 其实就是要关闭一个连接前的各种资源销毁处理，这一步的销毁主要是删除当前TcpServer中保存的信息*/
            void remove_inactive_connection_in_loop(const ConnectionPtr &conn) {
                {
                    /*这里考虑移除操作有可能在多个eventloop线程中同时进行，而unordered_map是非线程安全的，因此进行加锁保护移除过程*/
                    std::unique_lock<std::mutex> lock(_mutex);
                    /*释放conn保存在tcp_server中的shared_ptr,这样等最终外界所有的shared_ptr处理完销毁后续事项后被销毁，计数为0，则conn被释放*/
                    _connections.erase(conn->id());
                }
                DBG_LOG("delete conn from tcpserver:%p", conn.get());
                /*当前的处理完毕后，向处理队列中压入连接销毁的处理函数,其实本身就是在eventloop线程中调用，因此conn_destroyed是直接被执行的*/
                EventLoop *ioloop = conn->get_loop();
                ioloop->run_in_loop(std::bind(&Connection::shutdown, conn));
            }
        private:
            /*这两个成员，是每次建立连接，Connection初始化后，设置的连接建立完成回调，和消息处理回调函数*/
            /*同时，这两个函数是自己可以编写默认的处理函数，或者由外界具体协议服务来设定*/
            ConnectionComplateCallback _connected_callback;
            ConnectionMsgCallback _on_message_callback;
        private:
            int _port;
            uint64_t _next_id;
            EventLoop *_base_loop;
            Acceptor *_acceptor;
            EventLoopThreadPool *_thread_pool;
            std::mutex _mutex;
            std::unordered_map<uint64_t, ConnectionPtr> _connections;
    };

    /*------------------------------------------------------Buffer-----------------------------------------------------------------------*/

    Buffer::Buffer(size_t init_size = DEFAULT_BUFFER_SIZE):_reader_idx(0), _writer_idx(0), _buffer(init_size) {}
    char *Buffer::begin() { return &*_buffer.begin(); }
    /*返回当前读取位置之前空闲空间大小*/
    size_t Buffer::prepend_size() { return _reader_idx; }
    /*查看剩余空间是否足够len大小，不够就resize空间大小，够了就将当前读位置的数据移动到空间起始处*/
    void Buffer::make_space(size_t len) {
        if(writeable_size() + prepend_size() < len) {
            _buffer.resize(_writer_idx + len);
        }else {
            size_t sz = readable_size();
            std::copy(begin() + _reader_idx, begin() + _writer_idx, begin());
            _reader_idx = 0;
            _writer_idx = _reader_idx + sz;
        }
    }
    /*将当前写位置向后移动*/
    void Buffer::has_written(size_t len) {
        assert(len <= writeable_size());
        _writer_idx += len;
    }
    /*确保剩余可写空间大小足够len长度，不够的话就make_space*/
    void Buffer::ensure_write_able_size(size_t len) {
        if (len > writeable_size()) {
            make_space(len);
        }
        assert(writeable_size() >= len);
    }
    void Buffer::retrieve_all() {
        _reader_idx = 0;
        _writer_idx = 0;
    }
    /*返回当前写位置地址*/
    char *Buffer::begin_write() { return begin() + _writer_idx; }
    /*返回当前可读数据大小*/
    size_t Buffer::readable_size() { return _writer_idx - _reader_idx; }
    /*返回当前可读数据后的可写空间大小*/
    size_t Buffer::writeable_size() { return _buffer.size() - _writer_idx; }
    /*返回当前读取位置地址*/
    const char *Buffer::peek() { return begin() + _reader_idx; }
    /*返回当前行的结束地址*/
    const char *Buffer::find_eol() {
        const void *pos = memchr(peek(), '\n', readable_size());
        return (const char *)pos;
    }
    /*当前读位置向后偏移指定长度*/
    void Buffer::retrieve(size_t size) {
        assert(size <= readable_size());
        if (size < readable_size()) _reader_idx += size;
        else retrieve_all();
    }
    /*将指定长度的可读数据以string取出，并向后偏移读位置*/
    std::string Buffer::retrieve_as_string(size_t len) {
        assert(len <= readable_size());
        std::string res(peek(), len);
        retrieve(len);
        return res;
    }
    std::string Buffer::retrieve_all_as_string() {
        return retrieve_as_string(readable_size());
    }
    std::string Buffer::getline() {
        const char *pos = find_eol();
        /*没有找到换行，表示这不是一个完整的行，则返回一个空string*/
        if (pos == NULL) return "";
        /*+1是为了把换行\n字符也取出来*/
        return retrieve_as_string(pos - peek() + 1);
    }
    /*将数据写入缓冲区中*/
    void Buffer::append(const char *data, size_t len) {
        const char *d = (const char *)data;
        if (len > writeable_size()) {
            ensure_write_able_size(len);
        }
        std::copy(d, d + len, begin_write());
        has_written(len);
    }
    /*------------------------------------------------------Buffer-----------------------------------------------------------------------*/
    /*------------------------------------------------------Socket-----------------------------------------------------------------------*/
    Socket::Socket():_sockfd(-1) {}
    Socket::Socket(int fd):_sockfd(fd) {}
    Socket::~Socket() { close(); }
    int Socket::fd() { return _sockfd; }
    void Socket::set_reuse() {
        int opt = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(int));
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(int));
    }
    void Socket::set_nonblock() {
        int flag = fcntl(_sockfd, F_GETFL, 0);
        fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
    }
    bool Socket::socket() {
        _sockfd = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (_sockfd < 0) {
            ERR_LOG("socket error");
            return false;
        }
        return true;
    }
    bool Socket::bind(const std::string &ip, int port) {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = ::bind(_sockfd, (struct sockaddr*)&addr, len);
        if (ret < 0) {
            ERR_LOG("bind error");
            return false;
        }
        return true;
    }
    bool Socket::connect(const std::string &ip, int port) {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = ::connect(_sockfd, (struct sockaddr*)&addr, len);
        if (ret < 0) {
            ERR_LOG("connect error");
            return false;
        }
        return true;
    }
    bool Socket::listen(int backlog = MAX_LISTEN) {
        int ret = ::listen(_sockfd, backlog);
        if (ret < 0) {
            ERR_LOG("listen error");
            return false;
        }
        return true;
    }
    int Socket::accept() {
        int newfd = ::accept(_sockfd, NULL, 0);
        if (newfd < 0) {
            //ERR_LOG("accept error");
            return -1;
        }
        DBG_LOG("new socket:%d", newfd);
        return newfd;
    }
    bool Socket::create_listener(int port, const std::string &ip = "0.0.0.0") {
        assert(socket());
        set_nonblock();
        set_reuse();
        assert(bind(ip, port));
        assert(listen());
        return true;
    }
    bool Socket::create_client(const std::string &srvip, int srvport) {
        assert(socket());
        assert(connect(srvip, srvport));
        return true;
    }
    ssize_t Socket::recv(void *buf, size_t len, int flag = 0) {
        int ret = ::recv(_sockfd, buf, len, flag);
        if (ret < 0) {
            if (errno == EAGAIN || errno == EINTR) {
                return 0;
            }
            ERR_LOG("recv error");
            return -1;
        }else if (ret == 0) {
            INF_LOG("peer shutdown!");
            return -1;
        }
        return ret;
    }
    ssize_t Socket::nonblock_recv(void *buf, size_t len) { 
        return recv(buf, len, MSG_DONTWAIT); 
    }
    ssize_t Socket::send(const void *data, size_t len, int flag = 0) {
        int ret = ::send(_sockfd, data, len, flag);
        if (ret < 0) {
            if (errno == EAGAIN || errno == EINTR) {
                return 0;
            }
            ERR_LOG("send error!");
            return -1;
        }
        return ret;
    }
    ssize_t Socket::nonblock_send(const void *buf, size_t len) { 
        return send(buf, len, MSG_DONTWAIT); 
    }
    void Socket::close() { 
        if (_sockfd >= 0) {
            DBG_LOG("close sockfd:%d", _sockfd);
            ::close(_sockfd); 
            _sockfd = -1; 
        } 
    }
    /*------------------------------------------------------Socket-----------------------------------------------------------------------*/
    /*------------------------------------------------------Channel-----------------------------------------------------------------------*/
    Channel::Channel(EventLoop *loop, int fd):_loop(loop), _fd(fd), _events(0), _revents(0){}
    int Channel::fd() { return _fd; }
    uint32_t Channel::events() { return _events; };
    void Channel::set_event_callback(const EventCallback &cb) { _event_callback = cb; }
    void Channel::set_read_callback(const EventCallback &cb) { _read_callback = cb; }
    void Channel::set_write_callback(const EventCallback &cb) { _write_callback = cb; }
    void Channel::set_error_callback(const EventCallback &cb) { _error_callback = cb; }
    void Channel::set_close_callback(const EventCallback &cb) { _close_callback = cb; }
    void Channel::remove() { return _loop->remove_channel(this); }
    void Channel::update() { return _loop->update_channel(this); }
    void Channel::set_revents(uint32_t events) { _revents = events; }
    void Channel::enable_read() { _events |= EPOLLIN; update(); }
    void Channel::disable_read() { _events &= ~EPOLLIN; update(); }
    void Channel::enable_write() { _events |= EPOLLOUT; update(); }
    void Channel::disable_write() { _events &= ~EPOLLOUT; update(); }
    void Channel::disable_all() { _events = 0; update(); }
    bool Channel::readable() { return _events & EPOLLIN; }
    bool Channel::writeable() { return _events & EPOLLOUT; }
    EventLoop *Channel::owner_loop() { return _loop; }
    void Channel::handle_event() {
        if (_revents & (EPOLLIN | EPOLLPRI | EPOLLRDHUP)) {
            if (_read_callback) _read_callback();
        }
        if (_revents & EPOLLOUT) {
            if (_write_callback) _write_callback();
        }
        if (_revents & EPOLLERR) {
            if (_error_callback) _error_callback();
        }
        if (_revents & EPOLLHUP) {
            if (_close_callback) _close_callback();
        }
        if (_event_callback) _event_callback();
        return ;
    }
    /*------------------------------------------------------Channel-----------------------------------------------------------------------*/

    /*------------------------------------------------------Connection-----------------------------------------------------------------------*/
    Connection::Connection(EventLoop *loop, int fd, uint64_t id):
        _id(id), 
        _sock(new Socket(fd)), 
        _loop(loop), 
        _channel(new Channel(loop, fd)), 
        _state(CONNECTING) {
        _sock->set_nonblock();
        _channel->set_event_callback(std::bind(&Connection::handle_event, this));
        _channel->set_read_callback(std::bind(&Connection::handle_read, this));
        _channel->set_write_callback(std::bind(&Connection::handle_write, this));
        _channel->set_error_callback(std::bind(&Connection::handle_error, this));
        _channel->set_close_callback(std::bind(&Connection::handle_close, this));
    }
    void Connection::establisehd_in_loop() {
        assert(this->_state == CONNECTING);
        this->_state = CONNECTED;
        /*将当前channel添加到_loop的事件监控中，并启动可读事件监控*/
        _channel->enable_read();
        /*调用外界设置的连接成功回调函数--这个函数后边主要是设置context上下文操作*/
        if (_connected_callback) _connected_callback(shared_from_this());
    } 
    void Connection::shutdown_in_loop() {
        /*这个接口一定是对外的close接口被调用时压入处理队列的，那么当时的状态就一定是CONNECTED，断言失败大概率是外界重复shutdown了*/
        if (_state == DISCONNECTED) { return ; }
        _state = DISCONNECTING;
        _channel->disable_write();
        /*如果要关闭，又没有待发送的数据，则直接关闭套接字，内存资源等定时器时间到了进行释放*/
        if (_out_buffer.writeable_size() == 0) {
            return handle_close();
        }
        /*如果有数据待发送，又要关闭连接，则启动可写事件，写完数据后再关闭*/
        if (_channel->writeable() == false) {
            _channel->enable_write();
        }
    }
    void Connection::send_in_loop(const void *msg, size_t len) {
        /*连接处于实际要释放关闭的状态，则什么都不需要做了*/
        if (_state == DISCONNECTED) return;
        /*如果当前发送缓冲区中没有数据，就直接开始写， 如果有数据就把数据加入缓冲区*/
        int ret = 0, remain = len;
        if (_out_buffer.readable_size() == 0) {
            assert(_sock->fd() == _channel->fd());
            ret = _sock->nonblock_send(msg, len);
            if (ret < 0)  return handle_close();
            remain -= ret;
        }
        /*将剩余没有发送完的数据添加到发送缓冲区中*/
        _out_buffer.append((char*)msg + ret, remain);
        /*如果没有启动可写监控，则启动可写事件监控*/
        if (_channel->writeable() == false) {
            _channel->enable_write();
        }
        return ;
    }
    void Connection::handle_event() {
        if (_event_callback) _event_callback(shared_from_this());
    }
    /*读取所有数据到接收缓冲区中*/
    void Connection::handle_read(){
        char buf[65536];
        int ret;
        assert(_sock->fd() == _channel->fd());
        while((ret = _sock->nonblock_recv(buf, 65535)) > 0) {
            _in_buffer.append(buf, ret);
        }
        /*如果读取出错了，或者连接断开了，则使用shutdown关闭,不能直接handle_close关闭描述符， 因为这时候有可能缓冲区有数据待发送,或者有数据待处理*/
        if (ret < 0) { 
            /*有可能出现对方发送完数据后直接关闭，因此如果接收缓冲区有数据，那就调用处理函数进行处理*/
            if (_in_buffer.readable_size() > 0) {
                /*if (_message_callback) _message_callback(shared_from_this(), &_in_buffer);*/
                /*但是这么做就要求上层的_message_callback数据处理函数，必须从缓冲区全部取出去，要不然上层不取出数据，这里又不关闭就会不断有可读事件*/
                /*因此在这里不再监控可读事件，不触发事件则也不会刷新定时器任务，超时也就会关闭掉了*/
                _channel->disable_read();
            }else {
                return shutdown();
            }
        }
        /*拿到数据后调用外界传入的消息处理函数*/
        if (_message_callback) _message_callback(shared_from_this(), &_in_buffer);
    }
    void Connection::handle_write() {
        assert(_sock->fd() == _channel->fd());
        /*写入缓冲区有数据可读，表示的是有数据可写，则发送数据*/
        if (_out_buffer.readable_size() != 0){
            /*对发送缓冲区中的数据进行全部发送*/
            ssize_t ret = _sock->nonblock_send(_out_buffer.peek(), _out_buffer.readable_size());
            if (ret < 0) { return handle_close(); }
            /*发送缓冲区读位置向后偏移ret长度*/
            _out_buffer.retrieve(ret);
        }
        if (_out_buffer.readable_size() == 0) {
            /*若发送缓冲区中没有数据待发送，则解除写事件监控*/
            _channel->disable_write();
            /*若处于待关闭状态，则关闭释放资源*/
            if (_state == DISCONNECTING) {
                handle_close();
            }
        }
    }
    void Connection::handle_error() {
        /*目前并未对错误事件进行特殊操作，所以暂时可以不管，统一按照关闭处理*/
        handle_close();
    }
    void Connection::handle_close() {
        /*解除所有的事件监控*/
        if (this->_state != DISCONNECTED) {
            this->_state = DISCONNECTED;
            _channel->disable_all();
            /*这个主要是移除poller中保存的channel信息以及epoll中的事件监控*/
            _channel->remove();
            /*关闭描述符*/
            _sock->close();
        }
        if (_connected_callback) _connected_callback(shared_from_this());
    }


    EventLoop *Connection::get_loop() { return _loop; }
    Connection::~Connection() { DBG_LOG("release conn source:%p", this);  }
    int Connection::id() { return _id; } 
    int Connection::fd() { assert(_sock->fd() == _channel->fd());  return _sock->fd(); }
    bool Connection::connected() { return this->_state == CONNECTED;}
    Buffer *Connection::input_buffer() { return &_in_buffer; }
    Buffer *Connection::output_buffer() { return &_out_buffer; }
    void Connection::set_context(const Any &context) { _context = context; }
    Any *Connection::get_context() { return &_context; }
    void Connection::set_event_callback(const ConnectionEventCallback &cb) { _event_callback = cb; }
    void Connection::set_connected_callback(const ConnectionComplateCallback &cb) { _connected_callback = cb; }
    void Connection::set_on_message_callback(const ConnectionMsgCallback &cb) { _message_callback = cb; }
    void Connection::set_close_callback(const ConnectionCloseCallback &cb) { _close_callback = cb; }
    void Connection::send(const void *msg, size_t len) {
        if (_loop->is_in_loop_thread()) {
            send_in_loop(msg, len);
        }else {
            _loop->run_in_loop(std::bind(&Connection::send_in_loop, this, msg, len));
        }
    }
    void Connection::shutdown() {
        if (_loop->is_in_loop_thread()) {
            shutdown_in_loop();
        }else {
            _loop->run_in_loop(std::bind(&Connection::shutdown_in_loop, this));
        }
    }
    /*这个函数是提供给TcpServer中使用，新连接创建成功后压入eventloop的连接完成后的处理函数，主要功能是设置Connection状态，以及添加可读事件监控，然后调用外部设置的连接完成回调*/
    void Connection::conn_established() {
        if (_loop->is_in_loop_thread()) {
            establisehd_in_loop();
        }else {
            _loop->run_in_loop(std::bind(&Connection::establisehd_in_loop, this));
        }
        return ;
    }
    /*------------------------------------------------------Connection-----------------------------------------------------------------------*/

    /*------------------------------------------------------Poller-----------------------------------------------------------------------*/
    Poller::Poller(EventLoop *loop): _loop(loop), _epfd(-1) {
        _epfd = epoll_create(MAX_EPOLLEVENTS);
        if (_epfd < 0) {
            ERR_LOG("create epoll failed!!");
            abort();
        }
    }
    void Poller::update(int op, Channel *channel) {
        struct epoll_event ev;
        ev.events = channel->events();
        ev.data.fd = channel->fd();
        epoll_ctl(_epfd, op, ev.data.fd, &ev);
    }
    bool Poller::has_channel(Channel *channel) {
        int fd = channel->fd();
        auto it = _channels.find(fd);
        if (it == _channels.end()) {
            return false;
        }
        return true;
    }
    void Poller::update_channel(Channel *channel) {
        if (has_channel(channel)) {
            return update(EPOLL_CTL_MOD, channel);
        }
        _channels.insert(std::make_pair(channel->fd(), channel));
        return update(EPOLL_CTL_ADD, channel);
    }
    void Poller::remove_channel(Channel *channel) {
        /*这边后边还要修改，应该在没有事件后删除，要根据channel的状态来决定*/
        auto it = _channels.find(channel->fd());
        if (it != _channels.end()) {
            _channels.erase(it);
        }
        return update(EPOLL_CTL_DEL, channel);
    }
    bool Poller::poll(std::vector<Channel*> *active_channels, int timeout = -1) {
        active_channels->clear();
        std::chrono::high_resolution_clock::time_point start = std::chrono::high_resolution_clock::now();
        int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, timeout);
        std::chrono::high_resolution_clock::time_point end = std::chrono::high_resolution_clock::now();
        if (nfds < 0) {
            if (errno == EINTR) {
                return true;
            }
            ERR_LOG("epoll wait error!!");
            return false;
        }else if (nfds == 0) {
            return true;
        }
        for (int i = 0; i < nfds; i++) {
            int fd = _evs[i].data.fd;
            auto it = _channels.find(fd);
            assert(it != _channels.end());
            it->second->set_revents(_evs[i].events);
            active_channels->push_back(it->second);
        }
        auto t1 = std::chrono::duration_cast<std::chrono::duration<double>>(end - start);
        if (_loop->is_main())  INF_LOG("epoll wait: %lf", t1.count());
        return true;
    }
    /*------------------------------------------------------Poller-----------------------------------------------------------------------*/

    /*------------------------------------------------------TimerQueue-----------------------------------------------------------------------*/
    TimerQueue::TimerQueue(EventLoop *loop):
        _loop(loop), 
        _timer_fd(TimerQueue::create_timerfd()),
        _channel(new Channel(loop, _timer_fd)), 
        _capacity(DEFAULT_TIMEOUT), 
        _conns(DEFAULT_TIMEOUT){
        /*设置每次定时器时间到了的回调*/
        _channel->set_read_callback(std::bind(&TimerQueue::on_time, this));
        /*在eventloop线程中，将当前定时器关联_channel添加的可读事件监控中*/
        _channel->enable_read();
    }

    void TimerQueue::on_time() {
        read_timerfd();
        if (_conns.size() == _capacity) {
            _conns.pop_front();
        }
        _conns.push_back(Bucket());
    }
    void TimerQueue::erase(uint64_t id) {
        auto it = _timers.find(id);
        if (it != _timers.end()) {
            _timers.erase(it);
        }
    }
    void TimerQueue::timer_add(const OnTimerCallback &cb, uint64_t id) {
        INF_LOG("new timer task: %d", id);
        Timer *timer = new Timer(id);
        timer->set_on_time_callback(cb);
        timer->set_release_callback(std::bind(&TimerQueue::erase, this, id));
        PtrTimer pt(timer);
        _timers.insert(std::make_pair(id, WeakTimer(pt)));
        _conns.back().insert(pt);
    }
    void TimerQueue::timer_refresh(uint64_t id) {
        INF_LOG("refresh timer task: %d", id);
        auto it = _timers.find(id);
        if (it != _timers.end()) {
            _conns.back().insert(PtrTimer(it->second));
        }else {
            /*不可能存在这种情况呀, 添加的定时回调找不着*/
            ERR_LOG("%d timer task have not find!!", id);
            exit(-1);
        }
    }
    int TimerQueue::create_timerfd() {
        /*这里其实就是创建了一个每秒中会超时一次的定时器*/
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        struct itimerspec itm;
        itm.it_value.tv_sec = 1;//设置第一次超时的时间
        itm.it_value.tv_nsec = 0;
        itm.it_interval.tv_sec = 1;//第一次超时后，每隔多长时间超时
        itm.it_interval.tv_nsec = 0;
        timerfd_settime(timerfd, 0, &itm, NULL);//启动定时器
        return timerfd;
    }
    void TimerQueue::read_timerfd() {
        /*每次定时器时间到了，都会由系统向timerfd写入8字节数据，这个数据虽然当前没 用，但是必须取出来，不然会持续触发epoll就绪事件*/
        uint64_t times;
        int ret = ::read(_timer_fd, (char*)&times, sizeof(times));
        assert(ret == sizeof(times));
    }
    /*------------------------------------------------------TimerQueue-----------------------------------------------------------------------*/
    /*------------------------------------------------------Acceptor-----------------------------------------------------------------------*/
    Acceptor::Acceptor(EventLoop *loop, int port):_loop(loop)  {
        _sock.create_listener(port);
        _channel = new Channel(loop, _sock.fd());
        _channel->set_read_callback(std::bind(&Acceptor::read_handler, this));
        _channel->enable_read();
    }
    void Acceptor::set_new_conn_callback(const NewConnectionCallback &cb) {
        _new_conn_callback = cb;
    }
    /*这个函数是accept获取新连接的函数，到时候会将这个函数设置为channel的read_callback*/
    void Acceptor::read_handler() {
        int newfd;
        /*在一次事件触发中，将能获取的新连接全部都获取了，要不然每次重新监控可读事件，就绪后处理效率太低*/
        while((newfd = _sock.accept()) >= 0) {
            /*_new_conn_callback是在TcpServer构造中设置的回调，主要是为newfd创建Connection对象，并完成各项参数设置*/
            _new_conn_callback(newfd);
        }
        return ;
    }
    /*------------------------------------------------------Acceptor-----------------------------------------------------------------------*/
    /*------------------------------------------------------EventLoop-----------------------------------------------------------------------*/
    
    EventLoop::EventLoop(LoopType loop_type = MAIN_LOOP):
        _loop_type(loop_type),
        _thread_id(std::this_thread::get_id()),
        _weakup_fd(EventLoop::create_event_fd()),
        _poller(new Poller(this)),
        _weakup_channel(new Channel(this, _weakup_fd)),
        _timer_queue(new TimerQueue(this)){
        _weakup_channel->set_read_callback(std::bind(&EventLoop::weakup_channel_handler_read, this));
        _weakup_channel->enable_read();
    }
    int EventLoop::create_event_fd() {
        int fd = ::eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        assert(fd != -1);
        return fd;
    }
    void EventLoop::weakup() {
        uint64_t opts = 1;
        /*这里需要注意 eventfd创建的描述符每次只能进行8字节数据的传输进行事件通知，因此读写以8字节为单元*/
        int ret = ::write(_weakup_fd, &opts, sizeof(opts));
        if (ret != sizeof(opts)) {
            ERR_LOG("write weakup failed!! %d-%d-%s", errno, EBADF, strerror(errno));
            abort();
        }
    }
    void EventLoop::weakup_channel_handler_read() {
        uint64_t opts;
        int ret = ::read(_weakup_fd, &opts, sizeof(opts));
        if (ret != sizeof(opts)) {
            ERR_LOG("read weakup failed!!");
            abort();
        }
    }
    void EventLoop::do_pending_functor() {
        std::vector<Functor> functor;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            functor.swap(_pending_functor);
        }
        for (const Functor &f : functor) {
            f();
        }
    }
    
    void EventLoop::start() {
        INF_LOG("EventLoop start...");
        while(1) {
            std::chrono::high_resolution_clock::time_point start = std::chrono::high_resolution_clock::now();
            std::vector<Channel *> active_channels;
            bool ret = _poller->poll(&active_channels, DEFAULT_TIMEOUT * 1000);
            if (ret == false) {
                /*usleep(1000);*/
                continue;
            }
            std::chrono::high_resolution_clock::time_point mid = std::chrono::high_resolution_clock::now();
            for (Channel *channel : active_channels) {
                channel->handle_event();
            }
            do_pending_functor();
            std::chrono::high_resolution_clock::time_point end = std::chrono::high_resolution_clock::now();
            if (_loop_type == MAIN_LOOP) {
                auto t1 = std::chrono::duration_cast<std::chrono::duration<double>>(mid - start);
                auto t2 = std::chrono::duration_cast<std::chrono::duration<double>>(end - mid);
                INF_LOG("poll wait: %lf ;  handle: %lf ", t1.count(), t2.count());
                //std::cout << "poll wait:" <<  t1.count() << "; handle:" << t2.count() << std::endl;
            }
        }
        return ;
    }
    /*如果当前就是在eventloop线程中，则直接调用执行，否则就压入到任务队列中*/
    void EventLoop::run_in_loop(Functor cb) {
        if (is_in_loop_thread()) {
            return cb();
        }
        return queue_in_loop(cb);
    }
    /*将需要执行的函数压入eventloop线程任务队列中，然后如果当前不在eventloop线程中，则唤醒eventloop线程*/
    void EventLoop::queue_in_loop(Functor cb) {
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _pending_functor.push_back(cb);
        }
        if (is_in_loop_thread() == false) {
            weakup();
        }
        return ;
    }
    bool EventLoop::is_in_loop_thread() {
        return _thread_id == std::this_thread::get_id();
    }
    void EventLoop::update_channel(Channel *channel) {
        return _poller->update_channel(channel);
    }
    void EventLoop::remove_channel(Channel *channel) {
        return _poller->remove_channel(channel);
    }
    bool EventLoop::has_channel(Channel *channel) {
        return _poller->has_channel(channel);
    }
    void EventLoop::timer_task_enqueue(const OnTimerCallback &cb, uint64_t timer_id) {
        return _timer_queue->timer_add(cb, timer_id);
    }
    void EventLoop::timer_refresh(uint64_t timer_id) {
        return _timer_queue->timer_refresh(timer_id);
    }

    LoopThread::LoopThread():_loop(NULL), _thread(std::thread(&LoopThread::thread_entry, this)) {}
    void LoopThread::thread_entry() {
        /*在线程内部实例化eventloop，这样才能设置好eventloop对应的线程id*/
        EventLoop loop(SUB_LOOP);
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }
        return loop.start();
    }
    /*如果外界想要获取线程对应的eventloop，则阻塞至_loop不为NULL*/
    EventLoop *LoopThread::own_loop() { 
        EventLoop *loop = NULL;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&](){ return _loop != NULL; });
            loop = _loop;
        }
        return _loop;
    }


    
    EventLoopThreadPool::EventLoopThreadPool(EventLoop *baseloop):_base_loop(baseloop), _num_thread(0), _next(0) {}
    void EventLoopThreadPool::ser_thread_count(int num) {
        _num_thread = num;
    }
    EventLoop *EventLoopThreadPool::next_loop() {
        if (_num_thread == 0) {
            return _base_loop;
        }
        _next = (_next + 1) % _loops.size();
        return _loops[_next];
    }
    void EventLoopThreadPool::start() {
        if (_num_thread > 0) {
            _threads.resize(_num_thread);
            _loops.resize(_num_thread);
            for (int i = 0; i < _num_thread; i++) {
                _threads[i] = new LoopThread();
                _loops[i] = _threads[i]->own_loop();
            }
        }
        _base_loop->start();
    }
    /*------------------------------------------------------EventLoop-----------------------------------------------------------------------*/
}
class InitNetwork {
    public:
        InitNetwork() {
            signal(SIGPIPE, SIG_IGN);
            signal(SIGCHLD, SIG_IGN);
        }
};
static InitNetwork in;

#endif
